package com.imooc.coupon.service.impl;

import com.alibaba.fastjson.JSON;
import com.imooc.coupon.constant.CouponChannel;
import com.imooc.coupon.constant.CouponStatus;
import com.imooc.coupon.dao.CouponDao;
import com.imooc.coupon.entity.Coupon;
import com.imooc.coupon.service.IStreamConsumer;
import com.imooc.coupon.vo.CouponMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 * <h1>消息消费者接口实现</h1>
 * 核心思想: 是将 Cache 中的 Coupon 的状态变化同步到 DB 中
 * Created by Qinyi.
 */
@Slf4j
@Component
public class StreamConsumerImpl implements IStreamConsumer {

    @Autowired
    private CouponDao couponDao;

    /**
     * 消费优惠券 消息
     * @param message
     */
    @StreamListener(CouponChannel.COUPON_MESSAGE_INPUT)
    @Override
    public void consumeCouponMessage(Message<CouponMessage> message) {

        if (isEmptyMessage(message)) {
            return;
        }

        CouponMessage couponMessage = message.getPayload();

        log.info("Receive CouponMessage: {}", JSON.toJSONString(couponMessage));

        CouponStatus status = CouponStatus.of(couponMessage.getStatus());

        switch (status) {
            case USABLE:
                break;
            case USED:
                processUsedCoupons(couponMessage, status);
                break;
            case EXPIRED:
                processExpiredCoupons(couponMessage, status);
                break;
        }

    }

    /**
     * <h2>处理已使用的用户优惠券</h2>
     * */
    private void processUsedCoupons(CouponMessage couponMessage ,
                                    CouponStatus status) {
        // TODO 给用户发送短信
        processCouponsByStatus(couponMessage, status);
    }

    /**
     * <h2>处理过期的用户优惠券</h2>
     * */
    private void processExpiredCoupons(CouponMessage couponMessage,
                                       CouponStatus status) {
        // TODO 给用户发送推送
        processCouponsByStatus(couponMessage, status);
    }

    /**
     * 将优惠券状态更新到DB
     * @param couponMessage
     * @param status
     */
    private void processCouponsByStatus(CouponMessage couponMessage, CouponStatus status) {
        List<Coupon> coupons = couponDao.findAllById(couponMessage.getIds());
        if(CollectionUtils.isEmpty(coupons) || coupons.size() != couponMessage.getIds().size()){
            log.error("Can Not Find Right Coupon Info: {}",
                    JSON.toJSONString(couponMessage));
            // TODO 发送邮件
            return;
        }

        coupons.forEach(c->c.setStatus(status));
        log.info("CouponKafkaMessage Op Coupon Count: {}",couponDao.saveAll(coupons).size());
    }

    /**
     * 公用的空消息参数校验
     *
     * @param message
     * @return
     */
    protected boolean isEmptyMessage(Message message) {
        if (Objects.isNull(message)) {
            return true;
        }
        Object payload = message.getPayload();
        if (Objects.isNull(payload)) {
            return true;
        }
        return false;
    }
}
